package com.asiainfo.dacp.datastash.broker.core;

import com.asiainfo.dacp.datastash.broker.core.vo.StashUnit;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 *
 * 对于stashUnit的操作的封装，禁止外部直接操作的redisTemplate
 *
 * 不存在业务逻辑，只是允许数据操作
 *
 * @author MeiKefu
 * @date 2017-11-20
 */
@Component
public class StashUnitModel {

    static final String UNIT_CURRENT_BATCHNO_KEY = "datastash:unit_current_batchno";//当前状态

    static final String UNIT_LIST_KEY = "datastash:unit_list";//单元状态列表

    static final String UNIT_WAIT_QUEUE_KEY = "datastash:unit_wait_queue";//单元队列

    static final String UNIT_WAIT_QUEUE_LOCK = "datastash:unit_wait_queue:lock";//单元队列锁

    private static Logger LOG = LoggerFactory.getLogger(StashUnitModel.class);

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private StashTraceLogger tracelog;

    public void updateUnitBatchNo(String unitCode,String batchNo){
        redisTemplate.opsForHash().put(UNIT_CURRENT_BATCHNO_KEY,unitCode,batchNo);
    }

    public String getCurrentUnitBatchNo(String unitCode){
        return (String)redisTemplate.opsForHash().get(UNIT_CURRENT_BATCHNO_KEY,unitCode);
    }



    public void updateUnit(StashUnit stashUnit){
        redisTemplate.opsForHash().put(UNIT_LIST_KEY,stashUnit.getUnitProcId(),stashUnit);
    }

    public void delUnit(String unitProcId){
        redisTemplate.opsForHash().delete(UNIT_LIST_KEY,unitProcId);
    }
    /**
     * 如果不存在才放入
     */
    public void putUnitIfAbsent(StashUnit stashUnit){
        try{
            redisTemplate.opsForHash().putIfAbsent(UNIT_LIST_KEY,stashUnit.getUnitProcId(),stashUnit);
        }catch(Exception e){
            LOG.error("{}",e);
        }
    }

    public StashUnit getStashUnit(String unitProcId){
        StashUnit stashUnit = null;
        try{
            stashUnit = (StashUnit)redisTemplate.opsForHash().get(UNIT_LIST_KEY,unitProcId);
        }catch(Exception e){
            LOG.error("{}",e);
        }
        return stashUnit;
    }

    public List<StashUnit> getStashUnitList(){

        List<StashUnit> result = null;
        try{
            result = redisTemplate.opsForHash().values(UNIT_LIST_KEY);
        }catch(Exception e){
            LOG.warn("获取unitStatusList为空,{}",e);
        }
        return result;
    }




    /**
     * 获取一个等待判断依赖的任务
     * @return
     */
    public String fetchWaitingUnitProcId(){

        //1.判断共享锁，在更新桶的时候禁止去获取任务
        //if(isUnitWaitQueueLocked()){
            //return null;
        //}

        //2.返回任务
        String unitProcId = (String)redisTemplate.opsForList().leftPop(UNIT_WAIT_QUEUE_KEY);

        if(unitProcId!=null && unitProcId.length()>0){
            return unitProcId;
        }else{
            LOG.info("未获取到UNIT_WAIT_QUEUE的unitCode");
            return null;
        }
    }

    public void updateWaitQueue(final List<String> unitKeys){
//        redisTemplate.delete(UNIT_WAIT_QUEUE_KEY);
//        redisTemplate.opsForList().leftPushAll(UNIT_WAIT_QUEUE_KEY,unitKeys);
        final byte[] waitLockKey = redisTemplate.getKeySerializer().serialize(UNIT_WAIT_QUEUE_KEY);

        final byte[][] rawValues = new byte[unitKeys.size()][];
        int i = 0;
        for (String value : unitKeys) {
            rawValues[i++] =  redisTemplate.getValueSerializer().serialize(value);
        }

        RedisCallback<List<Object>> pipelineCallback = new RedisCallback<List<Object>>() {
            @Override
            public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
                //connection.del(waitLockKey);
                connection.rPush(waitLockKey,rawValues);
                return null;
            }
        };
        redisTemplate.executePipelined(pipelineCallback);
    }

    /**
     * 重新去排队
     */
    public void requeue(final StashUnit stashUnit){
//        redisTemplate.opsForHash().put(UNIT_STATUS_LIST_KEY,unitCode,stashUnit);
//        redisTemplate.opsForList().leftPush(UNIT_WAIT_QUEUE_KEY,unitCode);

        /*stashUnit.waiting();
        final byte[] statusListKey = redisTemplate.getKeySerializer().serialize(UNIT_LIST_KEY);
        final byte[] statusField = redisTemplate.getKeySerializer().serialize(stashUnit.getUnitProcId());
        final byte[] statusVal = redisTemplate.getHashValueSerializer().serialize(stashUnit);

        final byte[] waitQueueKey = redisTemplate.getKeySerializer().serialize(UNIT_WAIT_QUEUE_KEY);
        RedisCallback<List<Object>> pipelineCallback = new RedisCallback<List<Object>>() {
            @Override
            public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
                connection.hSet(statusListKey,statusField,statusVal);
                //connection.rPush(waitQueueKey,statusField);//右边入队
                return null;
            }
        };
        redisTemplate.executePipelined(pipelineCallback);*/
        redisTemplate.opsForList().rightPush(UNIT_WAIT_QUEUE_KEY, stashUnit.getUnitProcId());
    }

    /**
     * 等待队列是否被锁定
     * @return
     */
    public boolean isUnitWaitQueueLocked(){
        return redisTemplate.hasKey(UNIT_WAIT_QUEUE_LOCK);
    }

    /**
     * 锁定等待队列
     */
    public void lockUnitWaitQueue(){

//        redisTemplate.opsForValue().setIfAbsent(UNIT_WAIT_QUEUE_LOCK,"1");
//        redisTemplate.opsForValue().getOperations().expire(UNIT_WAIT_QUEUE_LOCK,1, TimeUnit.MINUTES);

        final byte[] waitLockKey = redisTemplate.getKeySerializer().serialize(UNIT_WAIT_QUEUE_LOCK);

        RedisCallback<List<Object>> pipelineCallback = new RedisCallback<List<Object>>() {
            @Override
            public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
                connection.setNX(waitLockKey,"1".getBytes());
                connection.expire(waitLockKey,60);
                return null;
            }
        };
        redisTemplate.executePipelined(pipelineCallback);
    }

    /**
     * 给等待队列解锁
     */
    public void unlockUnitWaitQueue(){
        redisTemplate.delete(UNIT_WAIT_QUEUE_LOCK);
    }
}
